package com.wxzhou.wangzhuan.scheduler;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.CollectionType;
import com.wxzhou.wangzhuan.entity.AdLog;
import com.wxzhou.wangzhuan.repository.AdRespository;
import io.quarkus.redis.datasource.ReactiveRedisDataSource;
import io.quarkus.redis.datasource.stream.ReactiveStreamCommands;
import io.quarkus.redis.datasource.stream.StreamMessage;
import io.quarkus.redis.datasource.stream.XReadArgs;
import io.quarkus.scheduler.Scheduled;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 定时消费广告日志，将广告日志批量持久化到数据库中
 * 使用普通XREAD模式而非消费者组模式，这样可以读取所有数据而不受确认状态影响
 */
@ApplicationScoped
public class RedisAdLogConsumer {

    private static final Logger LOG = Logger.getLogger(RedisAdLogConsumer.class);
    private static final int BATCH_SIZE = 2000;
    private static final String STREAM_KEY = "adlog:events";
    private static final String REDIS_STREAM_RESET_OFFSET = "REDIS_STREAM_RESET_OFFSET";
    private static final String CONSUMER_GROUP = "adlog-consumers";

    private final ReactiveStreamCommands<String, String, String> streamCommands;
    private final ObjectMapper objectMapper;
    private final CollectionType collectionType;
    private final AdRespository adRespository;

    @Inject
    public RedisAdLogConsumer(
            ReactiveRedisDataSource reactiveRedisDataSource,
            ObjectMapper objectMapper,
            AdRespository adRespository) {
        this.objectMapper = objectMapper;
        this.collectionType = objectMapper.getTypeFactory().constructCollectionType(List.class, AdLog.class);
        this.streamCommands = reactiveRedisDataSource.stream(String.class, String.class, String.class);
        this.adRespository = adRespository;

        // 使用普通读取模式，不需要创建消费者组
        LOG.debug("RedisAdLogConsumer已初始化，使用普通读取模式(XREAD)");
    }

//    @Scheduled(every = "10s", concurrentExecution = Scheduled.ConcurrentExecution.SKIP)
    public Uni<Void> consumeAdLogs() {
        return Uni.createFrom().voidItem()
                .invoke(() -> {
//                    LOG.debug("使用普通读取模式(XREAD)消费消息");
                })
                .chain(() -> {
                    // 获取消费的起始偏移量
                    String startId = getStartOffset();
//                    LOG.debugf("从偏移量消费消息: %s", startId);

                    // 使用正确的API调用方式
                    return streamCommands.xread(STREAM_KEY, startId, new XReadArgs().count(BATCH_SIZE).block(Duration.ZERO));
                })
                .chain(messages -> {
                    if (messages != null && !messages.isEmpty()) {
//                        LOG.infof("从Redis流接收到%d条广告日志事件", messages.size());
                        return processAdLogs(messages);
                    } else {
                        LOG.debug("没有新消息需要消费");
                        return Uni.createFrom().voidItem();
                    }
                })
                .onFailure().invoke(throwable -> {
                    LOG.error("从Redis流读取广告日志事件时出错", throwable);
                })
                .replaceWithVoid();
    }

    /**
     * 获取消费的起始偏移量
     * 默认使用">"只消费新消息，但可以通过环境变量修改以支持重新消费
     */
    private String getStartOffset() {
        // 可以通过环境变量来控制是否需要重新消费
        String resetOffset = System.getenv(REDIS_STREAM_RESET_OFFSET);
        if (resetOffset != null && !resetOffset.isEmpty()) {
            // 如果设置了环境变量，则根据其值决定起始偏移量
            if ("true".equalsIgnoreCase(resetOffset) || "0".equals(resetOffset)) {
                LOG.debug("Redis流将从头开始消费消息(0)");
                return "0";
            } else if (">".equals(resetOffset)) {
                LOG.debug("Redis流将只消费新消息(>");
                return ">";
            } else {
                // 支持指定具体的消息ID作为起始偏移量
                LOG.debugf("Redis流将从指定偏移量开始消费: %s", resetOffset);
                return resetOffset;
            }
        }

        // 默认行为：只消费新消息
        return "0";
    }

    private Uni<Void> processAdLogs(List<StreamMessage<String, String, String>> records) {
        return Uni.createFrom().voidItem()
                .invoke(() -> {
                    try {
                        // 存储解析成功的广告日志
                        List<AdLog> adLogs = new ArrayList<>();
                        Map<String, StreamMessage<String, String, String>> messageMap = new HashMap<>();

                        // 解析所有消息，每一批消息表示某个设备某一刻上报的广告数据含ECPM和用户行为数据。
                        for (StreamMessage<String, String, String> record : records) {
                            if (record == null) {
                                LOG.debugf("批次中接收到空记录");
                                continue;
                            }

                            // 获取消息 ID 和数据
                            String messageId = record.id();
                            Map<String, String> eventData = record.payload();

                            if (eventData == null || eventData.isEmpty()) {
                                LOG.warnf("消息ID: %s的事件数据为空或为null", messageId);
                                continue;
                            }

                            if (!eventData.containsKey("data")) {
                                LOG.warnf("消息ID: %s的事件数据中缺少必需的'data'字段", messageId);
                                continue;
                            }

                            try {
                                String eventJson = eventData.get("data");
                                if (eventJson == null || eventJson.isEmpty()) {
                                    LOG.warnf("消息ID: %s的事件JSON为空或为null", messageId);
                                    continue;
                                }

                                // 将事件数据反序列化为 AdLog 对象
                                List<AdLog> adLogList = objectMapper.readValue(eventJson, collectionType);
                                adLogs.addAll(adLogList);
                                messageMap.put(messageId, record);
                            } catch (Exception e) {
                                LOG.errorf("解析ID为: %s的广告日志事件时出错", messageId, e);
                            }
                        }

                        // 批量插入到数据库
                        if (!adLogs.isEmpty()) {
                            //TODO：异步执行数据库插入
                            adRespository.batchInsert(adLogs)
                                    .subscribe().with(
                                            rowCount -> {
                                                // 批量确认和删除消息，只有当影响行数大于0时才执行
                                                if (rowCount > 0) {
                                                    acknowledgeMessages(messageMap, rowCount);
                                                } else {
                                                    LOG.warn("没有广告日志事件被插入数据库或者插入失败，仍然删除该消息");
                                                    //TODO:好的做法是，把该数据写入到redis的set中，用于后续分页查询其中处理失败的记录。
                                                    acknowledgeMessages(messageMap, rowCount);
                                                }
                                            },
                                            throwable -> {
                                                LOG.error("将广告日志事件插入数据库时出错", throwable);
                                                // 记录失败日志，不确认消息，这样消息会在下次继续被消费
                                                logFailedMessages(messageMap);
                                            }
                                    );
                        }
                    } catch (Exception e) {
                        LOG.error("处理广告日志事件批次时出错", e);
                    }
                });
    }

    private void acknowledgeMessages(Map<String, StreamMessage<String, String, String>> messageMap, int insertedRowCount) {
        try {
            // 使用AtomicInteger来包装计数器，使其可以在lambda中安全修改
            final AtomicInteger processedCount = new AtomicInteger(0);
            final int finalInsertedRowCount = insertedRowCount;

            // 批量确认和删除消息
            for (String messageId : messageMap.keySet()) {
                try {
                    // 确认消息处理完成 (XACK)
                    streamCommands.xack(STREAM_KEY, CONSUMER_GROUP, messageId)
                            .subscribe().with(acknowledgeResult -> {
                                        LOG.debugf("确认广告日志事件结果: %s ---- acknowledgeResult = %s", messageId, acknowledgeResult);
                                        streamCommands.xdel(STREAM_KEY, messageId)
                                                .subscribe().with(deleteResult -> {
                                                            if (deleteResult > 0) {
                                                                LOG.infof("成功确认并删除广告日志事件: %s", messageId);
                                                                int currentCount = processedCount.incrementAndGet();
                                                                if (currentCount == messageMap.size()) {
                                                                    LOG.infof("批次处理完成: 插入%d个事件，确认并删除%d条消息",
                                                                            finalInsertedRowCount, messageMap.size());
                                                                }
                                                            } else {
                                                                LOG.warnf("删除广告日志事件失败: %s", messageId);
                                                            }
                                                        },
                                                        deleteThrowable -> {
                                                            LOG.error("删除ID为: %s的广告日志事件时出错", messageId, deleteThrowable);
                                                        }
                                                );
                                    },
                                    acknowledgeThrowable -> {
                                        LOG.error("确认ID为: %s的广告日志事件时出错", messageId, acknowledgeThrowable);
                                    }
                            );
                } catch (Exception e) {
                    LOG.error("确认并删除ID为: %s的广告日志事件时出错", messageId, e);
                }
            }
        } catch (Exception e) {
            LOG.error("acknowledgeMessages方法执行出错", e);
        }
    }

    private void logFailedMessages(Map<String, StreamMessage<String, String, String>> messageMap) {
        try {
            for (String messageId : messageMap.keySet()) {
                LOG.warnf("广告日志事件持久化失败: %s，消息将被重新处理", messageId);
            }
        } catch (Exception e) {
            LOG.error("logFailedMessages方法执行出错", e);
        }
    }

    public void createConsumerGroupIfNotExists() {
        try {
            // 尝试创建消费者组，如果已经存在会抛出异常，我们忽略这个异常 创建消费者组时指定的偏移量 $ 表示只读取创建后新添加的消息。而使用偏移量 0 则表示从流的最开始（第一个消息）读取所有数据
            streamCommands.xgroupCreate(STREAM_KEY, CONSUMER_GROUP, "0").subscribe().with(
                    result -> {
                        LOG.infof("创建消费者组成功: %s", CONSUMER_GROUP);
                    },
                    throwable -> {
                        // 消费者组可能已经存在，这是正常的
                        LOG.debugf("消费者组可能已存在: %s", CONSUMER_GROUP);
                    }
            );
        } catch (Exception e) {
            // 捕获所有其他异常
            LOG.error("创建消费者组时出错", e);
        }
    }

}